package com.bdi.mqtt_broker.mqtt_protocol;

import com.bdi.mqtt_broker.constant.Constants;
import com.bdi.mqtt_broker.model.SessionStore;
import com.bdi.mqtt_broker.model.SubscribeStore;
import com.bdi.mqtt_broker.service.ChannelManagerService;
import com.bdi.mqtt_broker.service.ISessionStoreService;
import com.bdi.mqtt_broker.service.ISubscribeStoreService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.ReflectionUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 * MQTT 客户端 发送Publish消息
 */
@Slf4j
public class Publish {

    private final ISessionStoreService sessionStoreService;

    private final ISubscribeStoreService subscribeStoreService;

    private final ChannelGroup channelGroup;

    private final ChannelManagerService channelManagerService;

    public Publish(ISessionStoreService sessionStoreService, ISubscribeStoreService subscribeStoreService,
                   ChannelGroup channelGroup, ChannelManagerService channelManagerService){
        this.sessionStoreService = sessionStoreService;
        this.subscribeStoreService = subscribeStoreService;
        this.channelGroup = channelGroup;
        this.channelManagerService = channelManagerService;
    }

    /**
     * !!!!!!!!!!!!!!!!!!!!!!!!!!!!<br/>
     *  只支持 Qos = 0 的消息发送<br/>
     * !!!!!!!!!!!!!!!!!!!!!!!!!!!!<br/>
     * @param channel -
     * @param msg -
     */
    public void processPublish(Channel channel, MqttPublishMessage msg) {

        String clientId = (String) channel.attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).get();
        // Qos = 0
        if (msg.fixedHeader().qosLevel() == MqttQoS.AT_MOST_ONCE) {
            byte[] messageBytes = new byte[msg.payload().readableBytes()];
            msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
            this.sendPublishMessage(msg.variableHeader().topicName(), msg.fixedHeader().qosLevel(), messageBytes, false, false, channel);
        }
        // QoS=1
        if (msg.fixedHeader().qosLevel() == MqttQoS.AT_LEAST_ONCE) {
            // TODO: Need Implementation
        }
        // QoS=2
        if (msg.fixedHeader().qosLevel() == MqttQoS.EXACTLY_ONCE) {
            // TODO: Need Implementation
        }
        // retain=1, 保留消息
        if (msg.fixedHeader().isRetain()) {
            // TODO: Need Implementation
        }


    }

    public void sendPublishMessage(String topic, MqttQoS mqttQoS, byte[] messageBytes, boolean retain, boolean dup, Channel currentChannel) {
        List<SubscribeStore> subscribeStores = subscribeStoreService.search(topic);
        subscribeStores.forEach(subscribeStore -> {
            if (sessionStoreService.containsKey(subscribeStore.getClientId())) {
                // 订阅者收到MQTT消息的QoS级别, 最终取决于发布消息的QoS和主题订阅的QoS
                MqttQoS respQoS = mqttQoS.value() > subscribeStore.getMqttQoS() ? MqttQoS.valueOf(subscribeStore.getMqttQoS()) : mqttQoS;
                if (respQoS == MqttQoS.AT_MOST_ONCE) {
                    MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
                            new MqttPublishVariableHeader(topic, 0), Unpooled.buffer().writeBytes(messageBytes));
                    log.debug("PUBLISH - clientId: {}, topic: {}, Qos: {}", subscribeStore.getClientId(), topic, respQoS.value());
                    SessionStore sessionStore = sessionStoreService.get(subscribeStore.getClientId());
                    ChannelId channelId = channelManagerService.get(sessionStore.getBrokerId() + "_" + sessionStore.getChannelId());
                    if (channelId != null) {
                        Channel channel = channelGroup.find(channelId);
                        if (channel != null && currentChannel != channel) {
                            channel.writeAndFlush(publishMessage);
                        }
                    }
                }

                if (respQoS == MqttQoS.AT_LEAST_ONCE) {
                    // TODO: Need Implementation
                }

                if (respQoS == MqttQoS.EXACTLY_ONCE) {
                     // TODO: Need Implementation
                }
            }
        });
    }



}
